package com.atguigu.flink.sql.wordcount;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/25 0025 17:55
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 *
 * SQL:
 *  *    sqlQuery() : 查询
 *  *    executeSql() : 执行SQL ，负责DDL
 */
public class Flink02_SQLWordCount {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        //将流环境转换成成表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //将流转换成表
        Table table = tableEnv.fromDataStream(stream);

        //基于SQL操作
        //select x , xx , xxx from tableName where xxxx
        //不起名直接使用
        //Table result = tableEnv.sqlQuery("select user , url , ts from " + table + " where user <> 'Tom'");

        //起名
        //tableEnv.createTemporaryView("t1" , table);

        tableEnv.createTemporaryView("t1",stream);

        Table result = tableEnv.sqlQuery("select user , url , ts from t1 where user <> 'Tom'");

        result.execute().print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }

}
